package org.geektimes.reactive.streams;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description:
 *
 * @Author: menglinggang
 * @Date: 2021-03-31
 * @Time: 3:23 下午
 */
public class DefaultPublisher<T> implements Publisher<T> {

    private List<Subscriber<T>> subscribers = new LinkedList<>();
    private List<T> events ;

    public DefaultPublisher(List<T> events) {
        this.events = events;
    }

    @Override
    public void subscribe(Subscriber<? super T> subscriber) {
        if (events == null || events.isEmpty()) {
            return;
        }
        DefaultSubscription<T> subscription = new DefaultSubscription(subscriber);
        subscription.request(events.size());
        subscription.onSubscribe(subscription);
        subscribers.add(subscription);

        events.stream().forEach(item -> {
            try {
                if (subscription.isCancel()){
                    System.err.println("本次数据发布已忽略，数据为：" + item);
                } else {
                    subscription.onNext(item);
                }

            } catch (Exception e) {
                subscription.onError(e);
            }
        });
        subscription.onComplete();
    }

    public static void main(String[] args) {
        DefaultPublisher publisher = new DefaultPublisher(Arrays.asList(1,2,3,4,5));
        Subscriber<Object> objectDefaultSubscriber = new DefaultSubscriber<>();
        publisher.subscribe(objectDefaultSubscriber);
    }

}
